package com.atguigu.api5

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.functions.TableAggregateFunction
import org.apache.flink.types.Row
import org.apache.flink.util.Collector

/**
 * @description: xxx
 * @time: 2020/8/3 17:23
 * @author: baojinlong
 **/
object TableAggregateFunction {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    //设置事件时间机制
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 从文本读取
    val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data")
    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading): Long = {
          t.timestamp * 1000
        }
      })

    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment, settings)

    // 将DataStream转成Table.两种写法都可以
    // val sensorTable = tableEnv.fromDataStream(dataStream, 'id, 'timestamp as 'ts, 'temperature, 'et.rowtime)
    val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'timestamp.rowtime as 'ts, 'temperature)

    // table api
    val top2Temp = new Top2Temp
    val resultTable: Table = sensorTable
      .groupBy('id)
      .flatAggregate(top2Temp('temperature) as('temp, 'rank))
      .select('id, 'temp, 'rank)

    resultTable.toRetractStream[Row].print("table function")

    environment.execute("xxx")
  }
}

// 定义一个类用来表示表聚合函数的状态
class Top2TempAcc {
  var highestTemp: Double = Double.MinValue
  var secondHighestTemp: Double = Double.MinValue
}

class Top2Temp extends TableAggregateFunction[(Double, Int), Top2TempAcc] {
  override def createAccumulator(): Top2TempAcc = new Top2TempAcc

  // 实现计算聚合结果函数的accumulate
  def accumulate(acc: Top2TempAcc, temp: Double): Unit = {
    // 判断当前温度值,是否比状态中的值大
    if (temp > acc.highestTemp) {
      // 如果比最高温度还高.排在第一,原来的第一顺到第二位
      acc.secondHighestTemp = acc.highestTemp
      acc.highestTemp = temp
    } else if (temp > acc.secondHighestTemp) {
      // 如果在最高和第二高之前,那么直接替换第二高温度
      acc.secondHighestTemp = temp
    }
  }

  // 实现一个输出结果的方法,最终处理完表中所有数据时候调用

  def emitValue(acc: Top2TempAcc, out: Collector[(Double, Int)]): Unit = {
    out.collect((acc.highestTemp, 1))
    out.collect((acc.secondHighestTemp, 2))
  }
}